package cn.ac.iscas.service.common;

import cn.ac.iscas.domain.job.Job;
import cn.ac.iscas.domain.job.JobLog;
import cn.ac.iscas.domain.job.JobStatus;
import cn.ac.iscas.domain.response.job.SubmitJobResponse;
import cn.ac.iscas.service.executor.flink.FlinkJobExecutor;
import com.iscas.datasong.lib.common.DataSongException;

import org.json.JSONException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

/**
 * Created by ISCAS on 2017/11/2.
 *
 * 数据处理任务执行器，后台实际执行的服务
 *
 */
@Component
public class JobExecutor implements IJobExecutor {
	@Autowired
	private FlinkJobExecutor flinkJobExecutor;
//	@Autowired
//	private LocalJobExecutor localJobExecutor;


	/**
	 * 获取任务执行状态，执行日志
	 * @param id
	 * @return 返回按照时间排序的状态信息，可能为空
	 * @throws DataSongException
	 */
	@Override
	public List<JobLog> getJobLog(String id) throws DataSongException {
//		String result = httpClientUtils.doGet(EtlConfigure.getEtlServiceStatusUrl(id));
//
//		ResponseEntity<List<JobStatusMsg>> response = DataSongJsonUtils.fromJson(result,new TypeReference<ResponseEntity<List<JobStatusMsg>>>(){});
//
//		if(response.getStatus() == 200){
//			return response.getData();
//		}else{
//			throw new DataSongException(Status.SERVER_ERROR, response.getMessage());
//		}
		return null;
	}

	@Override
	public boolean stop(String jobInnerId) {
		//return localJobExecutor.stop(jobInnerId);
		return flinkJobExecutor.stop(jobInnerId);
	}

	@Override
	public Map<String, JobStatus> getJobStatus(List<String> ids) {
//		Map<String, JobStatus> statusMap = new HashMap<>();
//		statusMap.put(ids.get(0), JobStatus.RUNNING);
//
//		return statusMap;
		//return localJobExecutor.getJobStatus(ids);
		return flinkJobExecutor.getJobStatus(ids);
	}

	/**
	 * 开始执行任务
	 * @param job 任务体
	 * @return
	 * @throws DataSongException
	 */
	@Override
	public SubmitJobResponse execute(Job job) throws DataSongException, JSONException {

		//return localJobExecutor.execute(job);
		return flinkJobExecutor.execute(job.get_id());

//		SubmitJobResponse response = new SubmitJobResponse();
//		response.setStatus(Status.OK.getValue());
//
//		return response;
	}


}
